#include "processing_queue.h"
#include "lock_guard.h"

#include <assert.h>

namespace tinySrv
{
  
  ProcessingQueue::ProcessingQueue(unsigned threadCount, IProcessingQueueCallback *callback)
    : IsRun(true)
    , Callback(callback)
  {
    assert(!!callback);
    ThreadPool NewThreads;
    try
    {
      for (int i = 0 ; i < threadCount ; ++i)
      {
        NewThreads.push_back(new Thread(Thread::ICallbackPtr(new 
          ICallbackImpl<ProcessingQueue>(*this, &ProcessingQueue::ThreadFunc))));
      }
    }
    catch (...)
    {
      IsRun = false;
      try
      {
        ProcessingConditional.AlarmForAll();
      }
      catch (std::exception const &)
      {
        // TODO: log
      }
      DestroyThreads(&NewThreads);
      throw;
    }
    std::swap(Threads, NewThreads);
  }
  
  ProcessingQueue::~ProcessingQueue()
  {
    IsRun = false;
    try
    {
      ProcessingConditional.AlarmForAll();
    }
    catch (std::exception const &)
    {
      // TODO: log
    }
    DestroyThreads(&Threads);
  }
  
  void ProcessingQueue::Push(IRefObjectPtr item)
  {
    LockGuard<Spinlock> Lock(ItemsGuard);
    Items.push(item);
    ProcessingConditional.AlarmForAll();
  }
  
  void ProcessingQueue::DestroyThreads(ThreadPool *threads) const
  {
    for (ThreadPool::iterator i = threads->begin() ; i != threads->end() ; ++i)
      delete *i;
  }
  
  void ProcessingQueue::ThreadFunc()
  {
    while (IsRun)
    {
      try
      {
        ProcessingConditional.Wait();
        if (!IsRun)
          return;
        while (IsRun)
        {
          IRefObjectPtr Item;
          {
            LockGuard<Spinlock> Lock(ItemsGuard);
            if (Items.empty())
              break;
            Item = Items.front();
            Items.pop();
          }
          Callback->ProcessItem(Item);
        }
      }
      catch (std::exception const &)
      {
        // TODO: log
      }
    }
  }

}
